import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.sql.SparkSession;
import org.apache.spark.util.LongAccumulator;
import scala.Tuple2;
import java.util.Collections;
import java.util.Iterator;
import java.util.regex.Pattern;

public class TradeDataAnalysisSpark {
    private static final Pattern DATE_PATTERN = Pattern.compile("^\\d{6}$");

    public static void main(String[] args) {
        SparkSession spark = SparkSession.builder()
                .appName("TradeDataAnalysisSpark")
                .getOrCreate();
        JavaSparkContext sc = new JavaSparkContext(spark.sparkContext());

        LongAccumulator insufficientFieldsCounter = sc.sc().longAccumulator("InsufficientFields");
        LongAccumulator emptyFieldCounter = sc.sc().longAccumulator("EmptyField");
        LongAccumulator invalidDateFmtCounter = sc.sc().longAccumulator("InvalidDateFormat");
        LongAccumulator invalidDateRangeCounter = sc.sc().longAccumulator("InvalidDateRange");
        LongAccumulator invalidTagCounter = sc.sc().longAccumulator("InvalidTag");
        LongAccumulator invalidNumberCounter = sc.sc().longAccumulator("InvalidNumber");

        JavaRDD<String> inputRDD = sc.textFile(args[0]);

        JavaPairRDD<String, Tuple2<String, Long>> cleanedRDD = inputRDD.flatMapToPair(line -> {
            String[] fields = line.split(",");
            if (fields.length < 8) {
                insufficientFieldsCounter.add(1);
                return Collections.emptyIterator();
            }

            String rawDate = fields[0].trim();
            String tag = fields[1].trim();
            String country = fields[2].trim();
            String amountStr = fields[7].trim();

            if (rawDate.isEmpty() || tag.isEmpty() || country.isEmpty() || amountStr.isEmpty()) {
                emptyFieldCounter.add(1);
                return Collections.emptyIterator();
            }

            if (!DATE_PATTERN.matcher(rawDate).matches()) {
                invalidDateFmtCounter.add(1);
                return Collections.emptyIterator();
            }

            try {
                int year = Integer.parseInt(rawDate.substring(0, 4));
                int month = Integer.parseInt(rawDate.substring(4, 6));
                if (year < 1988 || year > 2020 || month < 1 || month > 12) {
                    invalidDateRangeCounter.add(1);
                    return Collections.emptyIterator();
                }
                String formattedDate = String.format("%d-%02d", year, month);

                if (!"1".equals(tag) && !"2".equals(tag)) {
                    invalidTagCounter.add(1);
                    return Collections.emptyIterator();
                }

                long amount = Long.parseLong(amountStr);
                String key = formattedDate + "," + country;
                return Collections.singletonIterator(new Tuple2<>(key, new Tuple2<>(tag, amount)));
            } catch (NumberFormatException e) {
                invalidNumberCounter.add(1);
                return Collections.emptyIterator();
            }
        });

        JavaPairRDD<String, Tuple2<Long, Long>> aggregatedRDD = cleanedRDD
                .groupByKey()
                .mapValues(values -> {
                    long export = 0;
                    long importAmt = 0;
                    for (Tuple2<String, Long> value : values) {
                        if ("1".equals(value._1())) {
                            export += value._2();
                        } else {
                            importAmt += value._2();
                        }
                    }
                    return new Tuple2<>(export, importAmt);
                });

        JavaRDD<String> resultRDD = aggregatedRDD.map(tuple -> {
            String[] keyParts = tuple._1().split(",");
            return keyParts[0] + "," + keyParts[1] + "," + tuple._2()._1() + "," + tuple._2()._2();
        });

        resultRDD.saveAsTextFile(args[1]);

        System.out.println("=== Data Error Statistics ===");
        System.out.println("InsufficientFields: " + insufficientFieldsCounter.value());
        System.out.println("EmptyField: " + emptyFieldCounter.value());
        System.out.println("InvalidDateFormat: " + invalidDateFmtCounter.value());
        System.out.println("InvalidDateRange: " + invalidDateRangeCounter.value());
        System.out.println("InvalidTag: " + invalidTagCounter.value());
        System.out.println("InvalidNumber: " + invalidNumberCounter.value());

        sc.stop();
    }
}
    